package com.netty.service;

import com.netty.constant.Constants;
import com.netty.helper.ChannelGroupHelper;
import com.netty.model.MessageWrapper;
import com.netty.model.SocketSession;
import com.netty.session.SocketSessionManager;
import com.netty.util.DwrUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;

/**
 * @author suzhen
 * @create 2018/7/13
 */
@Slf4j
@Service
public class ConnectServiceImpl implements ConnectService {

    @Resource
    private SocketSessionManager socketSessionManager;

    @Resource
    private MessageService messageService;

    @Override
    public void heartbeatToClient(ChannelHandlerContext handlerContext, MessageWrapper wrapper) {
        //设置心跳响应时间
        handlerContext.channel().attr(Constants.SessionConfig.SERVER_SESSION_HEARTBEAT).set(System.currentTimeMillis());
    }


    @Override
    public void pushMessage(ChannelHandlerContext ctx, MessageWrapper wrapper) {
        log.debug("新版信息发送");
        String sessionId = wrapper.getSessionId();
        if (!socketSessionManager.exist(sessionId)) {
            connect(ctx, wrapper);
        }

        /*
         * 服务器集群时，可以在此
         * 判断当前session是否连接于本台服务器，如果是，继续往下走，如果不是，将此消息发往当前session连接的服务器并 return
         * if(session!=null&&!session.isLocalhost()){//判断当前session是否连接于本台服务器，如不是
         * //发往目标服务器处理
         * return;
         * }
         */
        SocketSession session = socketSessionManager.getSession(sessionId);
        if (session != null) {
            //具体信息发送在session管理里面,发送是通过dwr推送
            ChannelGroupHelper.broadcast(wrapper.getBody());
            //boolean result = session.write(wrapper.getBody());
            //log.debug("session 信息发送结果: {}", result);
        }
    }

    @Override
    public void pushMessage(ChannelHandlerContext ctx, String sessionId, MessageWrapper wrapper) {
        if (!socketSessionManager.exist(sessionId)) {
            connect(ctx, wrapper);
        }
        //判断是不是无效用户回复
        //判断非机器人回复时验证
        if (!sessionId.equals(Constants.ImserverConfig.REBOT_SESSIONID)) {
            SocketSession session = socketSessionManager.getSession(sessionId);
            if (session == null) {
                throw new RuntimeException(String.format("session %s is not exist.", sessionId));
            }
            //返回给自己一份
            session.write(wrapper.getBody());
        }
        try {
            ///取得接收人 给接收人写入消息
            SocketSession responseSession = socketSessionManager.getSession(wrapper.getAcceptSessionId());
            if (responseSession != null && responseSession.isConnected()) {
                boolean result = responseSession.write(wrapper.getBody());
                if (result) {
                    messageService.saveOnlineMessageToDb(wrapper);
                } else {
                    messageService.saveOfflineMessageToDb(wrapper);
                }
            } else {
                messageService.saveOfflineMessageToDb(wrapper);
            }
        } catch (Exception e) {
            log.error("connector send occur PushException.", e);
            throw new RuntimeException(e.getCause());
        }


    }

    @Override
    public void pushGroupMessage(MessageWrapper wrapper) {
        ChannelGroupHelper.broadcast(wrapper.getBody());
        DwrUtil.sedMessageToAll(wrapper);
        //保存到数据库
        messageService.saveOnlineMessageToDb(wrapper);
    }

    @Override
    public boolean validateSession(MessageWrapper wrapper) {
        try {
            return socketSessionManager.exist(wrapper.getSessionId());
        } catch (Exception e) {
            log.error("connector validateSession Exception!", e);
            throw new RuntimeException(e.getCause());
        }
    }

    @Override
    public void close(ChannelHandlerContext handlerContext, MessageWrapper wrapper) {
        String sessionId = getChannelSessionId(handlerContext);
        if (StringUtils.isNotBlank(sessionId)) {
            close(handlerContext);
            log.warn("connector close channel sessionId -> " + sessionId + ", ctx -> " + handlerContext.toString());
        }
    }

    @Override
    public void close(String sessionId) {
        try {
            SocketSession session = socketSessionManager.getSession(sessionId);
            if (session != null) {
                socketSessionManager.removeSession(sessionId);
                List<Channel> list = session.getSessionAll();
                for (Channel ch : list) {
                    ChannelGroupHelper.remove(ch);
                }
                log.info("connector close sessionId -> " + sessionId + " success ");
            }
        } catch (Exception e) {
            log.error("connector close sessionId -->" + sessionId + "  Exception.", e);
            throw new RuntimeException(e.getCause());
        }
    }

    @Override
    public void close(ChannelHandlerContext handlerContext) {
        String sessionId = getChannelSessionId(handlerContext);
        try {
            String nid = handlerContext.channel().id().asShortText();
            SocketSession session = socketSessionManager.getSession(sessionId);
            if (session != null) {
                socketSessionManager.removeSession(sessionId, nid);
                ChannelGroupHelper.remove(handlerContext.channel());
                log.info("connector close sessionId -> " + sessionId + " success ");
            }
        } catch (Exception e) {
            log.error("connector close sessionId -->" + sessionId + "  Exception.", e);
            throw new RuntimeException(e.getCause());
        }
    }

    @Override
    public void connect(ChannelHandlerContext ctx, MessageWrapper wrapper) {
        try {
            String sessionId = wrapper.getSessionId();
            String sessionId0 = getChannelSessionId(ctx);
            //当sessionID存在或者相等  视为同一用户重新连接
            if (StringUtils.isNotEmpty(sessionId0) || sessionId.equals(sessionId0)) {
                log.info("connector reconnect sessionId -> " + sessionId + ", ctx -> " + ctx.toString());
                pushMessage(ctx, messageService.getReConnectionStateMsg(sessionId0));
            } else {
                log.info("connector connect sessionId -> " + sessionId + ", sessionId0 -> " + sessionId0 + ", ctx -> " + ctx.toString());
                socketSessionManager.createSession(wrapper, ctx);
                setChannelSessionId(ctx, sessionId);
                log.info("create channel attr sessionId " + sessionId + " successful, ctx -> " + ctx.toString());
            }
        } catch (Exception e) {
            log.error("connector connect  Exception.", e);
        }
    }

    @Override
    public boolean exist(String sessionId) {
        return socketSessionManager.exist(sessionId);
    }


    @Override
    public String getChannelSessionId(ChannelHandlerContext ctx) {
        return ctx.channel().attr(Constants.SessionConfig.SERVER_SESSION_ID).get();
    }

    private void setChannelSessionId(ChannelHandlerContext ctx, String sessionId) {
        ctx.channel().attr(Constants.SessionConfig.SERVER_SESSION_ID).set(sessionId);
    }

}
